package us.luosl.hadoop.mr.join;

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 两份数据中，如果有一份数据比较小，小数据全部加载到内存，按关键字建立索引。
 * 大数据文件作为map的输入文件，对map()函数每一对输入，都能够方便地和已加载到内存的小数据进行连接。
 * 把连接结果按key输出，经过shuffle阶段，reduce端得到的就是已经按key分组的，并且连接好了的数据。
 * 这种方法，要使用hadoop中的DistributedCache把小数据分布到各个计算节点，
 * 每个map节点都要把小数据库加载到内存，按关键字建立索引。
 * 这种方法有明显的局限性：有一份数据比较小，在map端，能够把它加载到内存，并进行join操作。
 * Created by luosl on 2018/9/3.
 */
public class MapSideJoin {

    public static class MapSideJoinMapper extends Mapper<Object, Text, NullWritable, JoinWritable> {

        private Map<Integer, JoinWritable> smallTableCache = new HashMap<Integer, JoinWritable>();

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            String charset = context.getConfiguration().get("smallTableCache.charset", "UTF-8");
            URI[] cfs = context.getCacheFiles();
            List<String> lines = FileUtils.readLines(new File(cfs[0].getPath()), charset);
            for(String line:lines){
                String[] arr = line.split(",");
                Integer userId = Integer.parseInt(arr[1]);
                JoinWritable value = new JoinWritable().setTableName("account")
                        .addData(new Text("accountId"), new IntWritable(Integer.parseInt(arr[0])))
                        .addData(new Text("userId"), new IntWritable(userId))
                        .addData(new Text("balance"), new DoubleWritable(Double.parseDouble(arr[2])));
                smallTableCache.put(userId, value);
            }
        }

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] arr = value.toString().split(",");
            if(arr.length > 1){
                int userId = Integer.parseInt(arr[0]);
                String nickName = arr[1];
                if(smallTableCache.containsKey(userId)){
                    JoinWritable writable = new JoinWritable().setTableName("user account")
                            .addData(new Text("userId"), new IntWritable(userId))
                            .addData(new Text("nickName"), new Text(nickName))
                            .setData(smallTableCache.get(userId).getData());
                    context.write(NullWritable.get(), writable);
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        File out = new File("out");
        if(out.exists()){
            FileUtils.forceDelete(out);
        }

        Configuration conf = new Configuration();
        // 关闭输出压缩
        conf.set("mapreduce.output.fileoutputformat.compress", "false");
        conf.set("smallTableCache.charset", "UTF-8");

        Job job = Job.getInstance(conf, "map side join");

        job.addCacheFile(new File(args[1]).toURI());

        job.setJarByClass(MapSideJoin.class);
        job.setMapperClass(MapSideJoin.MapSideJoinMapper.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(JoinWritable.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[2]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
